package etcd

import (
	"context"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/clientv3/concurrency"
	//"go.etcd.io/etcd/clientv3"
	"time"
)

/*
docker 安装：
拉取镜像：docker pull bitnami/etcd:latest
创建实例：
	创建网络：docker network create app-tier --driver bridge
	创建实例：docker run -d --name etcd-server --network app-tier  --publish 2379:2379 --publish 2380:2380 --env ALLOW_NONE_AUTHENTICATION=yes --env ETCD_ADVERTISE_CLIENT_URLS=http://etcd-server:2379  bitnami/etcd:latest
*/

type ETCDClient struct {
	*clientv3.Client
}

// NewETCDClient 新建一个etcd客户端
func NewETCDClient(info *ETCDConfig) (*ETCDClient, error) {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   info.Endpoints,
		DialTimeout: time.Duration(info.DialTimeout) * time.Second,
	})
	if err != nil {
		return nil, err
	} else {
		return &ETCDClient{cli}, nil
	}
}

// Close 关闭etcd，一般和new成对出现
func (e *ETCDClient) Close() error {
	return e.Client.Close()
}

// Put 简单的put
func (e *ETCDClient) Put(ctx context.Context, key, value string) error {
	_, err := e.Client.Put(ctx, key, value)
	return err
}

// PutWithLease 指定一个租期的put
func (e *ETCDClient) PutWithLease(ctx context.Context, key, value string, lease int64) error {
	resp, err := e.Client.Grant(context.TODO(), lease)
	if err != nil {
		return err
	}
	_, err = e.Client.Put(ctx, key, value, clientv3.WithLease(resp.ID))
	return err
}

// PutWithLeaseKeepAlive 指定一个租期，且自动续约的put 该方法也会阻塞，建议使用goruntine lease单位秒
func (e *ETCDClient) PutWithLeaseKeepAlive(ctx context.Context, key, value string, lease int64) error {
	resp, err := e.Client.Grant(context.TODO(), lease)
	if err != nil {
		return err
	}

	_, err = e.Client.Put(ctx, key, value, clientv3.WithLease(resp.ID))
	if err != nil {
		return err
	}

	_, err = e.Client.KeepAlive(context.TODO(), resp.ID)
	if err != nil {
		return err
	}
	// 这个结构带测试，不清楚为什么带上
	//for {
	//	<-ch
	//}
	return err
}

// Get 获取指定key全字匹配
func (e *ETCDClient) Get(ctx context.Context, key string) (map[string]string, error) {
	resp, err := e.Client.Get(ctx, key)
	if err != nil {
		return nil, err
	}
	result := make(map[string]string, 0)
	for _, ev := range resp.Kvs {
		result[string(ev.Key)] = string(ev.Value)
	}
	return result, nil
}

// GetPrefix 获取指定前缀的目录的所有key以及对应的值 类似有 WithLimit：限制数量；WithRev：特定版本；WithSort：排序
func (e *ETCDClient) GetPrefix(ctx context.Context, prefix string) (map[string]string, error) {
	resp, err := e.Client.Get(ctx, prefix, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	result := make(map[string]string, 0)
	for _, ev := range resp.Kvs {
		result[string(ev.Key)] = string(ev.Value)
	}
	return result, nil
}

// Del 删除指定key
func (e *ETCDClient) Del(ctx context.Context, key string) error {
	_, err := e.Client.Delete(ctx, key)
	return err
}

// Watch 注意调用该方法或阻塞，请考虑使用goruntine 异步调用callback
func (e *ETCDClient) Watch(ctx context.Context, key string, callback func(evType int32, key, value string) error) {
	rch := e.Client.Watch(ctx, key) // <-chan WatchResponse
	for wr := range rch {
		for _, ev := range wr.Events {
			go callback(int32(ev.Type), string(ev.Kv.Key), string(ev.Kv.Value))
		}
	}
}

// WatchPrefix 注意调用该方法或阻塞，请考虑使用goruntine 异步调用callback
func (e *ETCDClient) WatchPrefix(ctx context.Context, prefix string, callback func(evType int32, key, value string) error) {
	rch := e.Client.Watch(ctx, prefix, clientv3.WithPrefix()) // <-chan WatchResponse
	for wr := range rch {
		for _, ev := range wr.Events {
			go callback(int32(ev.Type), string(ev.Kv.Key), string(ev.Kv.Value))
		}
	}
}

// DistributedOperation 分布式操作
func (e *ETCDClient) DistributedOperation(mutexName string, callback func()) error {
	s1, err := concurrency.NewSession(e.Client)
	if err != nil {
		return err
	}
	defer s1.Close()

	m1 := concurrency.NewMutex(s1, mutexName)

	// 会话s1获取锁
	if err := m1.Lock(context.TODO()); err != nil {
		return err
	}

	// 执行callback
	callback()

	if err := m1.Unlock(context.TODO()); err != nil {
		return err
	}

	return nil
}

// DistributedOperationTTL 分布式操作,规定会话时间，做超时处理
func (e *ETCDClient) DistributedOperationTTL(mutexName string, callback func(), ttl int) error {
	s1, err := concurrency.NewSession(e.Client, concurrency.WithTTL(ttl))
	if err != nil {
		return err
	}
	defer s1.Close()

	m1 := concurrency.NewMutex(s1, mutexName)

	// 会话s1获取锁
	if err := m1.Lock(context.TODO()); err != nil {
		return err
	}

	// 执行callback
	callback()

	if err := m1.Unlock(context.TODO()); err != nil {
		return err
	}

	return nil
}

// DistributedOperationLease 分布式操作,规定会话时间，做超时处理
func (e *ETCDClient) DistributedOperationLease(mutexName string, callback func(), lease int64) error {
	resp, err := e.Client.Grant(context.TODO(), lease)
	if err != nil {
		return err
	}

	s1, err := concurrency.NewSession(e.Client, concurrency.WithLease(resp.ID))
	if err != nil {
		return err
	}
	defer s1.Close()

	m1 := concurrency.NewMutex(s1, mutexName)

	// 会话s1获取锁
	if err := m1.Lock(context.TODO()); err != nil {
		return err
	}

	// 执行callback
	callback()

	if err := m1.Unlock(context.TODO()); err != nil {
		return err
	}

	return nil
}
